package cc.ifok.lpc;

import cc.ifok.kit.Command;
import cc.ifok.kit.Message;
import cc.ifok.kit.Strings;
import net.schmizz.sshj.SSHClient;
import net.schmizz.sshj.connection.channel.Channel;
import net.schmizz.sshj.connection.channel.direct.Session;
import net.schmizz.sshj.transport.verification.PromiscuousVerifier;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedInputStream;
import java.io.BufferedReader;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

public final class LPCClient {
    private static final Logger LOG = LoggerFactory.getLogger(LPCClient.class);

    private final ConcurrentMap<String, Socks> socks = new ConcurrentHashMap<>();
    private final AtomicReference<SSHClient> client = new AtomicReference<>();
    private final Queue<ExecutorService> execs = new ConcurrentLinkedQueue<>();
    private final BlockingQueue<Socks> clears = new LinkedBlockingQueue<>();

    public LPCClient(){daemonSSHClient();startInDaemon();}

    public boolean send(String sid, String message) {
        Socks sock = Optional.ofNullable(socks.get(sid)).orElseGet(() -> {
            socks.put(sid, new Socks(sid));return socks.get(sid);
        });
        return Optional.ofNullable(sock).map(sck -> {
            if (sck.opened()) {
                if(sck.send(message)) {
                    return true;
                } else {
                    close(sck.sid.get());
                    return false;
                }
            } else {
                close(sck.sid.get());
                return false;
            }
        }).orElse(false);
    }

    public void close(String sid) {
        Optional.ofNullable(socks.remove(sid)).ifPresent(sock -> {
            try{clears.put(sock);}catch(Throwable e){/*ignore exception*/}
        });
    }

    public int counts(){return socks.size();}

    private void daemonSSHClient() {
        Optional.ofNullable(client.get()).ifPresent(ssh -> {
            try{ssh.disconnect();}catch(Throwable e){/*ignore exception*/}
        });
        try {client.set(new SSHClient());
            client.get().loadKnownHosts();
            client.get().setConnectTimeout(9000);
            client.get().addHostKeyVerifier(new PromiscuousVerifier());
        } catch (Throwable e) {e.printStackTrace();}
    }
    private void onMessage(String sid, Command cmd, String line) {
        try{Message.LPS.put(new Message(sid, cmd, line));}catch(Throwable e){/*ignore exception*/}
    }

    private ExecutorService takeThreadExecutor() {
        return Optional.ofNullable(execs.poll()).orElseGet(Executors::newSingleThreadExecutor);
    }
    private final class Socks {
        private final AtomicReference<ExecutorService> thread = new AtomicReference<>();
        private final AtomicReference<Session.Shell> shell = new AtomicReference<>();
        private final AtomicReference<Future<?>> future = new AtomicReference<>();
        private final AtomicReference<Session> session = new AtomicReference<>();
        private final AtomicReference<String> sid = new AtomicReference<>();
        private Socks(String sid) {
            try {
                session.set(client.get().startSession());
                shell.set(session.get().startShell());
                if (opened()) {thread.set(takeThreadExecutor());
                    final InputStream input = shell.get().getInputStream();
                    future.set(thread.get().submit(() -> {
                        InputStream buffer = new BufferedInputStream(input);
                        InputStreamReader isr = new InputStreamReader(buffer);
                        BufferedReader reader = new BufferedReader(isr);
                        try {String line;
                            while (null != (line = reader.readLine())) {
                                onMessage(sid, Command.MESSAGED, line);
                            }
                        }catch(Throwable e){/*ignore exception*/}finally {
                            try{buffer.close();isr.close();reader.close();input.close();close(sid);}catch(Throwable e){/*ignore*/}
                        }
                    }));
                }
            } catch (Throwable e) {/*ignore exception*/}
        }
        private boolean opened() {
            return Optional.ofNullable(session.get()).map(Channel::isOpen).orElse(false)
                   && Optional.ofNullable(shell.get()).map(Channel::isOpen).orElse(false);
        }
        private boolean send(String message) {
            return Optional.ofNullable(shell.get()).map(shell -> {
                try {
                    if (opened()) {
                        shell.getOutputStream().write(Strings.bytes(message));
                        return true;
                    } else {
                        return false;
                    }
                } catch (Throwable e) {/*ignore exception*/return false;}
            }).orElse(false);
        }
        private void destroy() {
            // close task
            Optional.ofNullable(future.get()).ifPresent(fte -> {
                if (!fte.isDone() && !fte.isCancelled()) {
                    fte.cancel(true);
                }
            });
            // recycling thread
            Optional.ofNullable(thread.get()).ifPresent(exec -> {
                if (!exec.isShutdown()) {
                    execs.offer(exec);
                }
            });
            // close shell
            Optional.ofNullable(shell.get()).ifPresent(shl -> {
                try {
                    shl.close();
                } catch (Throwable e) {/*ignore exception*/}
            });
            // close session
            Optional.ofNullable(session.get()).ifPresent(sss -> {
                try {
                    sss.close();
                } catch (Throwable e) {/*ignore exception*/}
            });
            // on message closed
            onMessage(sid.get(), Command.CLOSED, Strings.EMPTY);
        }
    }
    private static final ScheduledExecutorService TIMER = Executors.newSingleThreadScheduledExecutor();
    private static final ExecutorService CLEANER = Executors.newSingleThreadExecutor();
    private void startInDaemon() {
        final List<String> aes = new ArrayList<>();
        TIMER.scheduleAtFixedRate(() -> {
            if (!client.get().isConnected()) {
                try {client.get().connect("127.0.0.1", 22222);
                } catch (Throwable e){daemonSSHClient();
                    try{client.get().connect("127.0.0.1", 22222);}catch(Throwable ex){/*ignore*/}
                }
            }
            if (client.get().isConnected() && !client.get().isAuthenticated()) {
                try {
                    client.get().authPassword("root", "1234567890!");aes.clear();
                } catch (Throwable e) {
                    if(aes.isEmpty()){aes.add(e.getMessage());LOG.error(e.getMessage());}
                }
            }
            if (client.get().isConnected() && client.get().isAuthenticated()) {
                try (Session session = client.get().startSession()) {
                    session.exec(("echo 'ping...' > /dev/null"));
                } catch (Throwable e) {/*ignore exception*/}
            }
            while(execs.size()>16){execs.poll().shutdownNow();}
        }, 128, 5120, TimeUnit.MILLISECONDS);

        CLEANER.submit(()->{while(Message.working()){try{clears.take().destroy();}catch(Throwable e){/*ignore*/}}});
    }
}
